/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal_fastdds/entity_impl/DalClientImpl.hpp"
#include "dal_fastdds/listener_impl/ClientRespReaderListenerImpl.hpp"
#include "dal_fastdds/utils/log.hpp"

using namespace eprosima::fastdds::dds;

ClientRespReaderListenerImpl::ClientRespReaderListenerImpl(DalClientImpl* dal_client_impl)
    : dal_client_impl_(dal_client_impl)
{
}

void ClientRespReaderListenerImpl::on_data_available(DataReader* reader)
{
    std::unique_lock<std::mutex> lock_mutex(on_new_resp_mutex_);
    if (on_resp_cb_) {
        uint64_t unread_resp_cnt = get_unread_responses();

        if (unread_resp_cnt > 0) {
            on_resp_cb_(user_data_);
        }
    }
}

void ClientRespReaderListenerImpl::on_subscription_matched(DataReader* reader,
                                                           const SubscriptionMatchedStatus& info)
{
    if (!dal_client_impl_)
        return;

    auto guid = eprosima::fastrtps::rtps::iHandle2GUID(info.last_publication_handle);
    if (info.current_count_change == 1) {
        matched_writers_.insert(guid);
    } else if (info.current_count_change == -1) {
        matched_writers_.erase(guid);
    } else {
        return;
    }

    dal_client_impl_->resp_reader_matched_count_.store(matched_writers_.size());
}

size_t ClientRespReaderListenerImpl::get_unread_responses()
{
    return dal_client_impl_->GetRespReader()->get_unread_count();
}

void ClientRespReaderListenerImpl::set_on_new_response_callback(dal_client_on_response_t callback,
                                                                const void* user_data)
{
    // handle responses before client is created.
    if (callback) {
        auto unread_resp_cnt = get_unread_responses();

        std::lock_guard<std::mutex> lock_mutex(on_new_resp_mutex_);
        if (unread_resp_cnt > 0) {
            callback(user_data);
        }

        user_data_ = user_data;
        on_resp_cb_ = callback;

        StatusMask mask = dal_client_impl_->GetRespReader()->get_status_mask();
        mask |= StatusMask::data_available();
        dal_client_impl_->GetRespReader()->set_listener(this, mask);
    } else {
        std::lock_guard<std::mutex> lock_mutex(on_new_resp_mutex_);

        StatusMask mask = dal_client_impl_->GetRespReader()->get_status_mask();
        mask &= ~StatusMask::data_available();
        dal_client_impl_->GetRespReader()->set_listener(this, mask);

        user_data_ = nullptr;
        on_resp_cb_ = nullptr;
    }
}